package com.disruptor.disruptorStudy.demo01;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkerPool;

public class Test02 {
    public static void main(String[] args) throws Exception {
        AtomicInteger code = new AtomicInteger();//生成累加id,线程安全的
        int BUFFER_size = 1024;
        int THREAD_NUM = 4;
        RingBuffer<Trade> ringbuffer = RingBuffer.createSingleProducer(
                () -> {
                    return new Trade();
                }, //event工厂
                BUFFER_size); //buffer大小
        //获取序列屏障(消费者与buffer的关系桥梁)
        SequenceBarrier sequenceBarrier = ringbuffer.newBarrier();
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUM);
        //创建workerpool,消息处理器
        @SuppressWarnings("unchecked")
        WorkerPool<Trade> workerPool = new WorkerPool<Trade>(
                ringbuffer,  //buffer
                sequenceBarrier,   //屏障
                new IgnoreExceptionHandler(), //异常
                //消费者:WorkHandler的实现类们( 可以有多个消费者 )
                (event) -> {
                    System.out.println("消费者1号拿到" + event.getId() + "号交易的价格:" + event.getPrice() + ",由" + event.getName() + "号生产者提供");
                },
                (event) -> {
                    System.out.println("消费者2号拿到" + event.getId() + "号交易的价格:" + event.getPrice() + ",由" + event.getName() + "号生产者提供");
                });
        workerPool.start(executor);//由线程池开启线程,消费者进行消费

        //生产者开始生产数据,所有消费者共同消费这些数据
        Future future1 = executor.submit(() -> {//new Callable<void>
            for (int i = 0; i < 10; i++) {
                ringbuffer.publishEvent((event, sequeue, buffer) -> {//EventTranslatorOneArg接口的实现类
                    event.setPrice(buffer.nextInt(1000));
                    event.setId(code.addAndGet(1) + "");//使用AtomicInteger生产累加id,并发可用
                    event.setName(Thread.currentThread().getName());
                }, new Random());//有参数,参数转化为buffer
            }
        });
        Thread.sleep(1000);
        //关闭
        workerPool.halt();
        executor.shutdown();
    }

}
